package com.hxl.alert.service;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

@Service
public class RocketMQAdminService {

    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Value("ALERT_CONSUMER_GROUP1")
    private String consumerGroup;

    @Value("ALERT_TOPIC")  // 指定要统计的目标 Topic
    private String topic;

    public long queryUnConsumedMessageCount() throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();
        try {
            mqAdminExt.setNamesrvAddr(nameServer);
            mqAdminExt.start();

            // 获取消费者组的消费统计信息
            ConsumeStats consumeStats = mqAdminExt.examineConsumeStats(consumerGroup);
            long totalUnread = 0;

            // 遍历所有队列，仅统计目标 Topic 的未消费消息
            for (MessageQueue queue : consumeStats.getOffsetTable().keySet()) {
                // 关键过滤：检查队列所属 Topic 是否匹配目标
                if (queue.getTopic().equals(topic)) {
                    long consumerOffset = consumeStats.getOffsetTable().get(queue).getConsumerOffset();
                    long brokerOffset = consumeStats.getOffsetTable().get(queue).getBrokerOffset();
                    totalUnread += (brokerOffset - consumerOffset);
                }
            }
            return totalUnread;
        } finally {
            mqAdminExt.shutdown();
        }
    }
}